/*-------------------------------------------------------------------------
 *
 * blvacuum.c
 *		Bloom VACUUM functions.
 *
 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  contrib/bloom/blvacuum.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/genam.h"
#include "bloom.h"
#include "catalog/storage.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "postmaster/autovacuum.h"
#include "storage/bufmgr.h"
#include "storage/indexfsm.h"
#include "storage/lmgr.h"


/*
 * 批量删除所有指向一组堆元组的索引条目。
 * 目标元组集通过一个回调例程指定，该例程指示
 * 是否任何给定的堆元组（通过 ItemPointer 标识）正在被删除。
 *
 * 结果：一个 palloc 的结构，包含 VACUUM 显示的统计信息。
 */
IndexBulkDeleteResult * blbulkdelete(IndexVacuumInfo *fc_info, IndexBulkDeleteResult *fc_stats,
			 IndexBulkDeleteCallback fc_callback, void *fc_callback_state)
{
	Relation	fc_index = fc_info->index;
	BlockNumber fc_blkno,
				fc_npages;
	FreeBlockNumberArray fc_notFullPage;
	int			fc_countPage = 0;
	BloomState	fc_state;
	Buffer		fc_buffer;
	Page		fc_page;
	BloomMetaPageData *fc_metaData;
	GenericXLogState *fc_gxlogState;

	if (fc_stats == NULL)
		fc_stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));

	initBloomState(&fc_state, fc_index);

	/*
	 * 迭代页面。我们不关心并发添加的页面，
	 * 因为它们无法包含要删除的元组。
	 */
	fc_npages = RelationGetNumberOfBlocks(fc_index);
	for (fc_blkno = BLOOM_HEAD_BLKNO; fc_blkno < fc_npages; fc_blkno++)
	{
		BloomTuple *fc_itup,
				   *fc_itupPtr,
				   *fc_itupEnd;

		vacuum_delay_point();

		fc_buffer = ReadBufferExtended(fc_index, MAIN_FORKNUM, fc_blkno,
									RBM_NORMAL, fc_info->strategy);

		LockBuffer(fc_buffer, BUFFER_LOCK_EXCLUSIVE);
		fc_gxlogState = GenericXLogStart(fc_index);
		fc_page = GenericXLogRegisterBuffer(fc_gxlogState, fc_buffer, 0);

		/* 忽略空的/已删除的页面，直到blvacuumcleanup() */
		if (PageIsNew(fc_page) || BloomPageIsDeleted(fc_page))
		{
			UnlockReleaseBuffer(fc_buffer);
			GenericXLogAbort(fc_gxlogState);
			continue;
		}

		/*
		 * 迭代元组。itup指向当前正在
		 * 扫描的元组，itupPtr指向下一个非删除元组的保存位置。
		 */
		fc_itup = fc_itupPtr = BloomPageGetTuple(&fc_state, fc_page, FirstOffsetNumber);
		fc_itupEnd = BloomPageGetTuple(&fc_state, fc_page,
									OffsetNumberNext(BloomPageGetMaxOffset(fc_page)));
		while (fc_itup < fc_itupEnd)
		{
			/* 我们必须删除这个元组吗？ */
			if (fc_callback(&fc_itup->heapPtr, fc_callback_state))
			{
				/* 是的；调整页面上将保留的元组计数 */
				BloomPageGetOpaque(fc_page)->maxoff--;
				fc_stats->tuples_removed += 1;
			}
			else
			{
				/* 否；复制到itupPtr++，但如果不需要则跳过复制 */
				if (fc_itupPtr != fc_itup)
					memmove((Pointer) fc_itupPtr, (Pointer) fc_itup,
							fc_state.sizeOfBloomTuple);
				fc_itupPtr = BloomPageGetNextTuple(&fc_state, fc_itupPtr);
			}

			fc_itup = BloomPageGetNextTuple(&fc_state, fc_itup);
		}

		/* 断言我们计数正确 */
		Assert(fc_itupPtr == BloomPageGetTuple(&fc_state, fc_page,
											OffsetNumberNext(BloomPageGetMaxOffset(fc_page))));

		/*
		 * 如果我们不会将页面标记为
		 * 删除，并且上面有空闲空间，则将页面添加到新的notFullPage列表中
		 */
		if (BloomPageGetMaxOffset(fc_page) != 0 &&
			BloomPageGetFreeSpace(&fc_state, fc_page) >= fc_state.sizeOfBloomTuple &&
			fc_countPage < BloomMetaBlockN)
			fc_notFullPage[fc_countPage++] = fc_blkno;

		/* 我们删除了什么吗？ */
		if (fc_itupPtr != fc_itup)
		{
			/* 现在是空页面吗？ */
			if (BloomPageGetMaxOffset(fc_page) == 0)
				BloomPageSetDeleted(fc_page);
			/* 调整pd_lower */
			((PageHeader) fc_page)->pd_lower = (Pointer) fc_itupPtr - fc_page;
			/* 完成WAL-日志记录 */
			GenericXLogFinish(fc_gxlogState);
		}
		else
		{
			/* 没有改变任何内容：中止WAL-日志记录 */
			GenericXLogAbort(fc_gxlogState);
		}
		UnlockReleaseBuffer(fc_buffer);
	}

	/*
	 * 用我们找到的任何内容更新metapage的notFullPage列表。我们的
	 * 信息此时可能已经过时，但blinsert()将会处理这种情况。
	 */
	fc_buffer = ReadBuffer(fc_index, BLOOM_METAPAGE_BLKNO);
	LockBuffer(fc_buffer, BUFFER_LOCK_EXCLUSIVE);

	fc_gxlogState = GenericXLogStart(fc_index);
	fc_page = GenericXLogRegisterBuffer(fc_gxlogState, fc_buffer, 0);

	fc_metaData = BloomPageGetMeta(fc_page);
	memcpy(fc_metaData->notFullPage, fc_notFullPage, sizeof(BlockNumber) * fc_countPage);
	fc_metaData->nStart = 0;
	fc_metaData->nEnd = fc_countPage;

	GenericXLogFinish(fc_gxlogState);
	UnlockReleaseBuffer(fc_buffer);

	return fc_stats;
}

/*
 * 后VACUUM清理。
 *
 * 结果：一个palloc'd结构包含VACUUM显示的统计信息。
 */
IndexBulkDeleteResult * blvacuumcleanup(IndexVacuumInfo *fc_info, IndexBulkDeleteResult *fc_stats)
{
	Relation	fc_index = fc_info->index;
	BlockNumber fc_npages,
				fc_blkno;

	if (fc_info->analyze_only)
		return fc_stats;

	if (fc_stats == NULL)
		fc_stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));

	/*
	 * 迭代页面：将已删除的页面插入FSM并收集
	 * 统计信息。
	 */
	fc_npages = RelationGetNumberOfBlocks(fc_index);
	fc_stats->num_pages = fc_npages;
	fc_stats->pages_free = 0;
	fc_stats->num_index_tuples = 0;
	for (fc_blkno = BLOOM_HEAD_BLKNO; fc_blkno < fc_npages; fc_blkno++)
	{
		Buffer		fc_buffer;
		Page		fc_page;

		vacuum_delay_point();

		fc_buffer = ReadBufferExtended(fc_index, MAIN_FORKNUM, fc_blkno,
									RBM_NORMAL, fc_info->strategy);
		LockBuffer(fc_buffer, BUFFER_LOCK_SHARE);
		fc_page = (Page) BufferGetPage(fc_buffer);

		if (PageIsNew(fc_page) || BloomPageIsDeleted(fc_page))
		{
			RecordFreeIndexPage(fc_index, fc_blkno);
			fc_stats->pages_free++;
		}
		else
		{
			fc_stats->num_index_tuples += BloomPageGetMaxOffset(fc_page);
		}

		UnlockReleaseBuffer(fc_buffer);
	}

	IndexFreeSpaceMapVacuum(fc_info->index);

	return fc_stats;
}
